package com.navinfo.platform.crosscity.service;


import com.navinfo.platform.common.service.IDataHandleService;
import com.navinfo.platform.common.service.impl.CommonService;
import org.apache.commons.io.FileUtils;
import org.apache.spark.api.java.function.FilterFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.io.File;
import java.io.IOException;
import java.net.URISyntaxException;
import java.util.*;


public enum CrossCityDataHandleService implements IDataHandleService {
    INS;

    @Override
    public void handle(List<Object> obj, Map<String, String> configMap, SparkSession spark) {
        Dataset<Row> dataSet = castValue(obj.get(0));

        List<String> tids = null;
        try {
            File file = FileUtils.getFile( new File(CrossCityDataHandleService.class.getProtectionDomain().getCodeSource().getLocation().toURI().getPath()).getParentFile(), "config/crosscity/tid.txt") ;
            tids = FileUtils.readLines(file);
        } catch (URISyntaxException |IOException e) {
            e.printStackTrace();
            System.out.println("加载tid失败...");
            throw new RuntimeException("加载tid失败...");
        }

        if(null==tids || tids.size()<=0){
            throw new RuntimeException("加载tid失败...");
        }

        List<String> finalTids = tids;
        Map<String, Long> stringLongMap = dataSet.filter(new FilterFunction<Row>() {
            @Override
            public boolean call(Row value) throws Exception {
                return finalTids.contains(value.getAs(0)+"");
            }
        }).javaRDD().flatMap(row -> {
            String cross_cities = row.getAs("cross_cities");
            String[] split = null;
            if (null != cross_cities && cross_cities.length() > 0) {
                split = cross_cities.split(",");
            }

            return Arrays.asList(split).iterator();
        }).countByValue();

        List<Map.Entry<String, Long>> list = new LinkedList<Map.Entry<String, Long>>(stringLongMap.entrySet());
        Collections.sort(list, (o1, o2) -> (o2.getValue()).compareTo(o1.getValue()));

        for(int i=0; i<list.size() && i<21; i++){
            System.out.println("top"+i+":"+list.get(i).getKey()+"---"+list.get(i).getValue());
        }
    }

}
